package com.ml.data;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.ml.enums.RequestType;
import com.ml.feign.OutDataRemote;
import com.ml.service.DatasService;
import com.ml.util.KlineDataUtils;
import com.ml.ws.model.DepthResp;
import com.ml.ws.model.PeriodLastKline;
import com.ml.ws.model.RequestParam;
import com.ml.ws.model.ResponseMessage;
import com.ml.ws.service.RequestFactory;
import com.ml.ws.service.RequestService;
import com.virgil.core.utils.GZipUtils;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;

/**
 * @author: Mr.D
 * @Date: 2020/8/10 17:11
 * @Description:
 *               <p>
 *               推送数据
 *               </p>
 */
@Component
public class PushData {
	public final static Logger LOGGER = LoggerFactory.getLogger(PushData.class);
	@Autowired
	DatasService datasService;
	@Autowired
	OutDataRemote outDataRemote;
	@Autowired(required = false)
	ConcurrentHashMap<Channel, RequestParam> depthMap;
	@Autowired(required = false)
	ConcurrentHashMap<Channel, RequestParam> klineMap;
	@Autowired(required = false)
	ConcurrentHashMap<Channel, RequestParam> tradeMap;
	@Autowired(required = false)
	ConcurrentHashMap<Channel, RequestParam> detailMap;

	/**
	 * @Description 推送最新K线数据
	 * @Author Mr.D
	 * @Date 2020/8/12 10:53
	 */
	@Scheduled(fixedDelay = 1000)
	public void pushKline() {
		// long start = System.currentTimeMillis();
		// 由于数据需要压缩流处理,当连接数多的情况下减少压缩处理,在此局部缓存一份
		Map<String, byte[]> datas = new HashMap<>();
		klineMap.forEach((k, v) -> {
			String symbol = v.getSymbol();
			String period = KlineDataUtils.getPeriod(v.getPeriod());
			// 从局部缓存拿压缩过的数据
			byte[] bytes = datas.get(symbol + "_" + period);
			if (bytes == null) {
				// 局部缓存没有则生成
				try {
					// 获取K线数据
					PeriodLastKline klineData = datasService.lastKline(symbol, period);
					// 封装消息体
					ResponseMessage message = new ResponseMessage("kline", klineData);
					// 获取压缩后的数据
					bytes = GZipUtils.compress(JSON.toJSONString(message));
					datas.put(symbol + "_" + period, bytes);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			// 转换成二进制数据
			ByteBuf buf = Unpooled.wrappedBuffer(bytes);
			k.writeAndFlush(new BinaryWebSocketFrame(buf));
		});
		// long end = System.currentTimeMillis();
		// LOGGER.info("花费时间:" + (end - start));
	}

	/**
	 * @Description 推送深度
	 * @Author Mr.D
	 * @Date 2020/8/12 10:53
	 */
	@Scheduled(fixedDelay = 1000)
	public void pushDepth() {
		outDataRemote.depth(1);
		// 由于数据需要转二进制压缩等处理,在此局部缓存一份
		Map<String, byte[]> datas = new HashMap<>();
		depthMap.forEach((k, v) -> {
			byte[] bytes = datas.get(v.getSymbol().toLowerCase() + "_" + v.getSize());
			if (bytes == null) {
				try {
					// 获取深度数据
					DepthResp depthResp = datasService.getDepth(v.getSymbol().toLowerCase(), v.getSize());
					// 封装消息体
					ResponseMessage message = new ResponseMessage("depth", depthResp);
					// 获取压缩后的数据
					bytes = GZipUtils.compress(JSON.toJSONString(message));
					datas.put(v.getSymbol().toLowerCase() + "_" + v.getSize(), bytes);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			// 转换成二进制数据
			ByteBuf buf = Unpooled.wrappedBuffer(bytes);
			k.writeAndFlush(new BinaryWebSocketFrame(buf));
		});

	}

	/**
	 * @Description 推送市场快照
	 * @Author Mr.D
	 * @Date 2020/8/12 10:54
	 */
	@Scheduled(fixedDelay = 1000)
	public void pushDetail() {
		try {
			// 封装消息体
			ResponseMessage message = new ResponseMessage("detail", datasService.getTicker());
			// 获取压缩后的数据
			byte[] bytes = GZipUtils.compress(JSON.toJSONString(message));
			detailMap.forEach((k, v) -> {
				// 转换成二进制数据
				ByteBuf buf = Unpooled.wrappedBuffer(bytes);
				k.writeAndFlush(new BinaryWebSocketFrame(buf));
			});
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void handlerReq(ChannelHandlerContext context, RequestParam param) {
		RequestService service = RequestFactory.getByRequestType(RequestType.of(param.getTopic()).get());
		service.pushReqData(context.channel(), param);
	}

	public void handlerSub(ChannelHandlerContext context, RequestParam param) {
		if ("kline".equals(param.getTopic())) {
			klineMap.put(context.channel(), param);
		}
		if ("depth".equals(param.getTopic())) {
			depthMap.put(context.channel(), param);
		}
		if ("trade".equals(param.getTopic())) {
			tradeMap.put(context.channel(), param);
		}
		if ("detail".equals(param.getTopic())) {
			detailMap.put(context.channel(), param);
		}
	}

	public void handlerUnsub(ChannelHandlerContext context, RequestParam param) {
		if ("kline".equals(param.getTopic())) {
			klineMap.remove(context.channel());
		}
		if ("depth".equals(param.getTopic())) {
			depthMap.remove(context.channel());
		}
		if ("trade".equals(param.getTopic())) {
			tradeMap.remove(context.channel());
		}
		if ("detail".equals(param.getTopic())) {
			detailMap.remove(context.channel());
		}
	}

}
